package yuekao1.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import yuekao1.entity.Tm4_1;
import yuekao1.entity.Tm4_2;
import yuekao1.util.ClickHouseUtil;
import yuekao1.util.KafkaUtil;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.HashSet;

public class ConsumeDwd {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // Kafka 订单明细主题读取数据，分组开窗聚合，统计各维度（省份、商品、用户）各窗口的订单数、订单金额、商品数量，补全维度信息，将数据写入 ClickHouse 交易域SKU粒度下单各窗口汇总表。
        //1）、编写Flink流式程序，从Kafka对象实时消费DWD层交易域下单事务事实表数据，设置允许乱序最大水位线为1秒，提取相关业务字段值，封装Java实体类对象；（5分）
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.kafkaSource("dwd_trade_orders"));
//        streamSource.print();
        SingleOutputStreamOperator<Tm4_1> tm4_1table = streamSource.map(new MapFunction<String, Tm4_1>() {
            @Override
            public Tm4_1 map(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                Integer id = jsonObject.getJSONObject("f0").getInteger("id");
                String province_id = jsonObject.getJSONObject("f0").getString("province_id");
                String name = jsonObject.getJSONObject("f2").getString("name");
                String sku_id = jsonObject.getJSONObject("f1").getString("sku_id");
                String sku_name = jsonObject.getJSONObject("f3").getString("sku_name");
                String user_id = jsonObject.getJSONObject("f0").getString("user_id");
                String user_name = jsonObject.getJSONObject("f4").getString("name");
                Double order_price = jsonObject.getJSONObject("f1").getDouble("order_price");
                Integer sku_num = jsonObject.getJSONObject("f1").getInteger("sku_num");
                String create_time = jsonObject.getJSONObject("f0").getString("create_time");
                return new Tm4_1(id, province_id, name, sku_id, sku_name, user_id, user_name, order_price, sku_num, create_time);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Tm4_1>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner((event, timestamp) -> {
                    try {
                        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
        tm4_1table.print();
        //2）、设置事件时间窗口为1天，每隔1秒触发窗口计算，实时累加统计各个维度的指标度量中（订单数据、订单金额、商品数据量）；（5分）
        //（省份、商品、用户）
        //    private String province_id;//省份
        //    private String name;//省份
        //    private String sku_id;//商品
        //    private String sku_name;//商品
        //    private String user_id;//用户
        //    private String user_name;//用户
        SingleOutputStreamOperator<Tm4_2> process = tm4_1table.keyBy(new KeySelector<Tm4_1, Tuple6<String, String, String, String, String, String>>() {
                    @Override
                    public Tuple6<String, String, String, String, String, String> getKey(Tm4_1 s) throws Exception {
                        return new Tuple6<>(s.getProvince_id(), s.getName(), s.getSku_id(), s.getSku_name(), s.getUser_id(), s.getUser_name());
                    }
                }).window(SlidingEventTimeWindows.of(Time.days(1), Time.minutes(1)))
                .process(new ProcessWindowFunction<Tm4_1, Tm4_2, Tuple6<String, String, String, String, String, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple6<String, String, String, String, String, String> one, ProcessWindowFunction<Tm4_1, Tm4_2, Tuple6<String, String, String, String, String, String>, TimeWindow>.Context context, Iterable<Tm4_1> iterable, Collector<Tm4_2> collector) throws Exception {
                        HashSet<Integer> set = new HashSet<>();
                        Double sumprice = 0.0;
                        Integer count = 0;
                        for (Tm4_1 o : iterable) {
                            set.add(o.getId());
                            sumprice += o.getOrder_price();
                            count += o.getSku_num();
                        }
                        String st = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getStart());
                        String ed = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getEnd());
                        collector.collect(new Tm4_2(st, ed, set.size(), sumprice, count, one.f1, one.f3, one.f5));
                    }
                });
        process.print();
        //3）、将上述交易域SKU粒度下单各窗口汇总结果数据，实时存储Clickhouse表，其中表的引擎ReplacingMergeTree主键相同时，更新字段值；（2分）
        process.addSink(new ClickHouseUtil());
        //备注：答题截图时，包括ClickHouse中创建表语句，查看表的数据等完整信息，否则0分处理。

        env.execute();
    }
}
